package cn.tecnova.cd

import java.sql.Timestamp

import cn.tecnova.utils.ConfigHandler
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{Dataset, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * description:
  * 221466
  **/
object TecnovaRecruit {
  Logger.getLogger("org").setLevel(Level.ERROR)

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setAppName(this.getClass.getSimpleName)
      //      .setMaster("local[*]")
      .set("es.index.auto.create", "true")
      .set("es.nodes", ConfigHandler.esNodes)
      .set("es.port", ConfigHandler.esPort)
      .set("es.nodes.wan.only", "true")

    val sc = new SparkContext(conf)
    val sqLContext = new SQLContext(sc)

    import sqLContext.implicits._
    val rowDF = sqLContext.read.jdbc("jdbc:mysql://www.slave4.com:3306/tec_base2", "company_recruit_information", ConfigHandler.props2)

    //导入写es相关的包
    import org.elasticsearch.spark.sql._

    val result: Dataset[TecnovaRecruit] = rowDF.map(row => {
      TecnovaRecruit(
        row.getAs[Int]("id"),
        row.getAs[String]("source"),
        row.getAs[String]("job_name"),
        row.getAs[String]("company_name"),
        row.getAs[String]("company_pay"),
        row.getAs[String]("monthly_pay"),
        row.getAs[String]("workplace"),
        row.getAs[String]("job_nature"),
        row.getAs[String]("experience"),
        row.getAs[String]("minimum_education"),
        row.getAs[Int]("recruitment"),
        row.getAs[String]("job_category"),
        row.getAs[Timestamp]("published_data"),
        row.getAs[String]("operating_duty"),
        row.getAs[String]("job_requirement"),
        row.getAs[String]("company_introduction"),
        row.getAs[String]("company_size"),
        row.getAs[String]("company_nature"),
        row.getAs[String]("company_industry"),
        row.getAs[String]("company_domain"),
        row.getAs[String]("company_adress"),
        row.getAs[Timestamp]("collect_data"),
        row.getAs[String]("job_url"),
        row.getAs[String]("key_no"),
        row.getAs[String]("com_res_rate")

      )
    })
    result.saveToEs("tecnova_recruit" + "/company_recruit")


    sc.stop()

  }

}

case class TecnovaRecruit(

                           var id: Long,
                           var source: String,
                           var jobName: String,
                           var companyName: String,
                           var companyPay: String,
                           var monthlyPay: String,
                           var workplace: String,
                           var jobNature: String,
                           var experience: String,
                           var minimumEducation: String,
                           var recruitment: Int,
                           var jobCategory: String,
                           var publishedData: Timestamp,
                           var operatingDuty: String,
                           var jobRequirement: String,
                           var companyIntroduction: String,
                           var companySize: String,
                           var companyNature: String,
                           var companyIndustry: String,
                           var companyDomain: String,
                           var companyAdress: String,
                           var collectData: Timestamp,
                           var jobUrl: String,
                           var keyNo: String,
                           var comResRate: String

                         )
